package com.audaque.springboot.foshanupload.web.flinkdemo.component;


import cn.hutool.json.JSONUtil;
import com.audaque.springboot.foshanupload.web.flinkdemo.configuration.KafkaConfig;
import com.audaque.springboot.foshanupload.web.flinkdemo.constants.KafkaTopic;
import com.audaque.springboot.foshanupload.web.flinkdemo.model.bo.TemperatureFlink;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.ExecutionException;

/**
 * @Auther: gzq
 * @Date: 2021/1/27 - 01 - 27 - 11:09
 * @Description: nevt
 */
@Component
public class Producer {
    @Resource
    private KafkaConfig kafkaConfig;
    private KafkaProducer<String, String> producer;

    private KafkaProducer<String, String> createProducer() {
        System.out.println("create producer");
        producer = new KafkaProducer<>(kafkaConfig.createPropsOfProducer());
        return producer;
    }

    private void closeKafka() {
        producer.close();
    }

    public void run() throws ExecutionException, InterruptedException {
        KafkaProducer<String, String> kafkaProducer = createProducer();
        System.out.println("begin write to kafka producer");
        HashMap<String, Integer> hashMap = new HashMap<>();
        hashMap.put("key", 1);
        new Thread(() -> {
            for (int i = 0; i < 100; i++) {
                TemperatureFlink temperatureFlink = new TemperatureFlink();
                temperatureFlink.setId("10003_2000005_30002_4000001_0");
                temperatureFlink.setTemp(i);
                temperatureFlink.setPres(20);
                temperatureFlink.setCTime(new Date().getTime());
                temperatureFlink.setUTime(new Date().getTime());
                String msg = JSONUtil.toJsonStr(temperatureFlink);
                kafkaProducer.send(new ProducerRecord<String, String>(KafkaTopic.kafkaTopicIn, msg));

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            }
        }).start();

    }
}
